/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2020-2022. All rights reserved.
 * Description: hvd adapter source file
 * Author: wangcimeng
 * Create: 2020-5-18
 */

#include "hvd_adapter.h"
#include "hvd_ops_kernel_info_store.h"
#include "runtime/kernel.h"
#include "sal_pub.h"
using namespace std;

namespace hccl {
HvdCbDataProcessFunc g_cbDataProcessFunc;    // 处理callback数据的函数指针
HvdAdapter g_hvdAdapterGlobal;               // 全局的HvdAdapter线程管理对象

void HcomRegHvdCallback(HvdCbDataProcessFunc fn)
{
    g_cbDataProcessFunc = fn;
}

// rts回调函数
void HvdOpCallbackFunc(void *fnData)
{
    HcclResult ret = g_cbDataProcessFunc(fnData);
    if (ret != HCCL_SUCCESS) {
        HCCL_WARNING("Horovod func process fndata fail");
    }
}

HvdAdapter::HvdAdapter()
{
}

HvdAdapter::~HvdAdapter()
{
}

HcclResult HvdAdapter::HvdAdapterInit(const rtStream_t stream, const s32 deviceID)
{
    HCCL_INFO("hvd_adapter into HvdAdapterInit");
    if (threadInited_) { // 线程已经被拉起，不再拉起线程
        std::vector<rtStream_t>::iterator streamSearch = std::find(threadStream_.begin(), threadStream_.end(), stream);
        if (streamSearch == threadStream_.end()) { // 被拉起的线程未和当前stream绑定
            rtError_t ret = rtSubscribeReport(threadId_, stream);
                CHK_PRT_RET(ret != RT_ERROR_NONE, HCCL_ERROR("[Init][HvdAdapter]errNo[0x%016llx] rtSubscribeReport"
                "fail,return[%d], para: threadId[%llu].",
                HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret, threadId_), HCCL_E_RUNTIME);
            threadStream_.push_back(stream);
        }
        return HCCL_SUCCESS;
    }

    hvdAdapterThread_.reset(new (std::nothrow) std::thread(&HvdAdapter::HvdAdapterThread, this, deviceID));
    while (!threadInited_) { // 等待线程被拉起
        SaluSleep(ONE_MILLISECOND_OF_USLEEP);
    }
    rtError_t ret = rtSubscribeReport(threadId_, stream);
    CHK_PRT_RET(ret != RT_ERROR_NONE, HCCL_ERROR("[Init][HvdAdapter]errNo[0x%016llx] rtSubscribeReport fail,"
        "return[%d], para: threadId[%llu].", HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret, threadId_), HCCL_E_RUNTIME);

    threadStream_.push_back(stream);
    HCCL_INFO("hvd_adapter into HvdAdapterInit end");
    return HCCL_SUCCESS;
}

HcclResult HvdAdapter::HvdAdapterThread(s32 deviceID)
{
    threadId_ = pthread_self();
    threadInited_ = true;
    shutDown_ = false;
    HCCL_INFO("hvd_adapter HvdAdapterThread inited threadId_[%llu], deviceID[%d]", threadId_, deviceID);
    rtError_t rtRet = rtSetDevice(deviceID);
    CHK_PRT_RET((rtRet != RT_ERROR_NONE), HCCL_ERROR("[HvdAdapter][Thread]errNo[0x%016llx] HvdAdapterThread"
        "rtSetDevice failed, return ret[%d]", HCCL_E_RUNTIME, rtRet), HCCL_E_RUNTIME);
    while (!shutDown_) {
        HCCL_DEBUG("rtProcessReport call start.");
        rtError_t ret = rtProcessReport(1000);    // 等待1000ms，等待callback函数的返回
        if (ret != RT_ERROR_NONE) {
            HCCL_INFO("errNo[0x%016llx] rtProcessReport fail, return[%d], expect[%d]", HCCL_ERROR_CODE(HCCL_E_RUNTIME),
                ret, RT_ERROR_NONE);
            SaluSleep(ONE_MILLISECOND_OF_USLEEP);
        }
        if (!threadInited_) {
            shutDown_ = true;
        }
        HCCL_DEBUG("hvd_adapter thread run loop once");
    }
    HCCL_INFO("hvd_adapter HvdAdapterThread end threadId_[%lld]", threadId_);
    return HCCL_SUCCESS;
}

HcclResult HvdAdapter::HvdAdapterDestroy()
{
    CHK_PRT_RET(threadInited_ != true, HCCL_ERROR("[Destroy][HvdAdapter]Hvd Adapter has not initialed now"),
        HCCL_E_INTERNAL);
    CHK_PRT_RET(shutDown_ == true, HCCL_ERROR("[Destroy][HvdAdapter]Hvd Adapter is in shutdown status now"),
        HCCL_E_INTERNAL);

    if (hvdAdapterThread_->joinable()) {
        threadInited_ = false;
        hvdAdapterThread_->join();
    } else {
        HCCL_ERROR("[Destroy][HvdAdapter]hvdAdapter thread is not joinable");
        return HCCL_E_INTERNAL;
    }
    hvdAdapterThread_.reset(nullptr);
    return HCCL_SUCCESS;
}
}
